/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.runners.gaia;



import com.bff.gaia.unified.runners.fnexecution.ServerFactory;

import com.bff.gaia.unified.runners.fnexecution.jobsubmission.JobInvoker;

import com.bff.gaia.unified.runners.fnexecution.jobsubmission.JobServerDriver;

import com.bff.gaia.unified.sdk.io.FileSystems;

import com.bff.gaia.unified.sdk.options.PipelineOptionsFactory;

import org.kohsuke.args4j.CmdLineException;

import org.kohsuke.args4j.CmdLineParser;

import org.kohsuke.args4j.Option;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;



import javax.annotation.Nullable;



/** Driver program that starts a job server for the Gaia runner. */

public class GaiaJobServerDriver extends JobServerDriver {



  private static final Logger LOG = LoggerFactory.getLogger(GaiaJobServerDriver.class);



  /** Gaia runner-specific Configuration for the jobServer. */

  public static class GaiaServerConfiguration extends ServerConfiguration {

    @Option(name = "--gaia-master-url", usage = "Gaia master url to submit job.")

    private String gaiaMasterUrl = "[auto]";



    String getGaiaMasterUrl() {

      return this.gaiaMasterUrl;

    }



    @Option(

        name = "--gaia-conf-dir",

        usage =

            "Directory containing Gaia YAML configuration files. "

                + "These properties will be set to all jobs submitted to Gaia and take precedence "

                + "over configurations in GAIA_CONF_DIR.")

    private String gaiaConfDir = null;



    @Nullable

    String getGaiaConfDir() {

      return gaiaConfDir;

    }

  }



  public static void main(String[] args) throws Exception {

    // TODO: Expose the fileSystem related options.

    // Register standard file systems.

    FileSystems.setDefaultPipelineOptions(PipelineOptionsFactory.create());

    fromParams(args).run();

  }



  private static void printUsage(CmdLineParser parser) {

    System.err.println(

        String.format("Usage: java %s arguments...", GaiaJobServerDriver.class.getSimpleName()));

    parser.printUsage(System.err);

    System.err.println();

  }



  public static GaiaJobServerDriver fromParams(String[] args) {

    GaiaServerConfiguration configuration = new GaiaServerConfiguration();

    CmdLineParser parser = new CmdLineParser(configuration);

    try {

      parser.parseArgument(args);

    } catch (CmdLineException e) {

      LOG.error("Unable to parse command line arguments.", e);

      printUsage(parser);

      throw new IllegalArgumentException("Unable to parse command line arguments.", e);

    }



    return fromConfig(configuration);

  }



  public static GaiaJobServerDriver fromConfig(GaiaServerConfiguration configuration) {

    return create(

        configuration,

        createJobServerFactory(configuration),

        createArtifactServerFactory(configuration));

  }



  public static GaiaJobServerDriver create(

      GaiaServerConfiguration configuration,

      ServerFactory jobServerFactory,

      ServerFactory artifactServerFactory) {

    return new GaiaJobServerDriver(configuration, jobServerFactory, artifactServerFactory);

  }



  private GaiaJobServerDriver(

      GaiaServerConfiguration configuration,

      ServerFactory jobServerFactory,

      ServerFactory artifactServerFactory) {

    super(configuration, jobServerFactory, artifactServerFactory);

  }



  @Override

  protected JobInvoker createJobInvoker() {

    return GaiaJobInvoker.create((GaiaServerConfiguration) configuration);

  }

}